package com.xux.listener;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.xux.enums.OperateTypes;
import com.xux.service.ICanalDataProcess;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import java.util.List;

@Slf4j
@Component
public class RocketListener {
    @Autowired
    private List<ICanalDataProcess> canalDataProcessList;
    @StreamListener("sp-tbl")
    public void orderSlic(Message<String> msg) {
        String payLoad = msg.getPayload();
        JSONObject payLoadJson = JSONUtil.parseObj(payLoad);
        String tableName = payLoadJson.getStr("table");
        String type = payLoadJson.getStr("type");
        OperateTypes existsType = OperateTypes.fromType(type);
        if (ObjectUtil.isNull(existsType)){
            log.error("["+type+"]类型不支持");
            return;
        }
        JSONArray dataArray = payLoadJson.getJSONArray("data");
        ICanalDataProcess canalDataProcess = canalDataProcessList.stream().filter(v -> {
            return v.support(tableName);
        }).findFirst().orElse(null);
        if (ObjectUtil.isNull(canalDataProcess)) {
            log.error("[op_order_slic]表不支持");
            return;
        }
        canalDataProcess.process(existsType,dataArray);
    }
}